package org.example.flink.sink;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormat;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.hadoopcompatibility.mapred.HadoopMapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class FileSinkT1 {

    private static final Logger log = LoggerFactory.getLogger(FileSinkT1.class);

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();


        Job job = Job.getInstance();
        HadoopInputFormat<LongWritable, Text> hadoopIF =
                new HadoopInputFormat<LongWritable, Text>(
                        new TextInputFormat(), LongWritable.class, Text.class, job
                );
        TextInputFormat.addInputPath(job, new Path("inputPath"));

// Read data using the Hadoop TextInputFormat.
        DataSet<Tuple2<LongWritable, Text>> text = env.createInput(hadoopIF);

        DataSet<Tuple2<Text, LongWritable>> result = text
                // use Hadoop Mapper (Tokenizer) as MapFunction
                .flatMap(new HadoopMapFunction<LongWritable, Text, Text, LongWritable>(
//                        new Tokenizer()
                ));

        env.execute("FileSinkT1");

    }


}
